package bma.common.netty;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelHandler;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.group.ChannelGroup;
import org.jboss.netty.channel.group.ChannelGroupFuture;
import org.jboss.netty.channel.group.ChannelGroupFutureListener;
import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import bma.common.langutil.ai.stack.AIStack;
import bma.common.langutil.ai.stack.AIStackSimple;
import bma.common.langutil.concurrent.ThreadPoolSquare;
import bma.common.langutil.core.ExceptionUtil;
import bma.common.langutil.io.HostPort;
import bma.common.langutil.log.LogPrinter.LEVEL;
import bma.common.langutil.pipeline.CommonPipelineBuilder;
import bma.common.langutil.pipeline.CommonPipelineBuilderChain;
import bma.common.langutil.spring.ServerBooter;
import bma.common.netty.handler.ChannelHandlerGroup;
import bma.common.netty.handler.ChannelHandlerLog;
import bma.common.netty.handler.ChannelHandlerLog.TYPE;
import bma.common.netty.handler.ChannelHandlerShowConnect;
import bma.common.netty.handler.NamedChannelHandler;

public class NettyServer implements ServerBooter {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(NettyServer.class);

	/**
	 * 配置数据
	 */
	protected String name;
	protected boolean ownBossExecutor;
	protected Executor bossExecutor;
	protected boolean ownWorkerExecutor;
	protected Executor workerExecutor;
	protected int worker = Runtime.getRuntime().availableProcessors() * 2;
	protected HostPort hostPort;
	protected CommonPipelineBuilder<ChannelHandler> pipelineBuilder;
	protected int traceBufferSize = 0;

	// socket config
	protected Boolean noDelay;

	/**
	 * 运行数据
	 */
	// ServerChannel
	protected transient ServerBootstrap bootstrap;
	protected transient Channel serverChannel;
	protected transient ChannelGroup channelGroup;

	public int getTraceBufferSize() {
		return traceBufferSize;
	}

	public void setTraceBufferSize(int traceBufferSize) {
		this.traceBufferSize = traceBufferSize;
	}

	public int getWorker() {
		return worker;
	}

	public void setWorker(int worker) {
		this.worker = worker;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public Executor getBossExecutor() {
		return bossExecutor;
	}

	public void setBossExecutor(Executor bossExecutor) {
		this.bossExecutor = bossExecutor;
	}

	public Executor getWorkerExecutor() {
		return workerExecutor;
	}

	public void setWorkerExecutor(Executor executor) {
		this.workerExecutor = executor;
	}

	public HostPort getHostPort() {
		return hostPort;
	}

	public void setHostPort(HostPort hostPort) {
		this.hostPort = hostPort;
	}

	public void setPort(int port) {
		this.hostPort = new HostPort(port);
	}

	public Boolean getNoDelay() {
		return noDelay;
	}

	public void setNoDelay(Boolean noDelay) {
		this.noDelay = noDelay;
	}

	public CommonPipelineBuilder<ChannelHandler> getPipelineBuilder() {
		return pipelineBuilder;
	}

	public void setPipelineBuilder(
			CommonPipelineBuilder<ChannelHandler> pipelineBuilder) {
		this.pipelineBuilder = pipelineBuilder;
	}

	public void addPipeline(NamedChannelHandler ch) {
		List<Object> l = new ArrayList<Object>(1);
		l.add(ch);
		setPipelines(l);
	}

	public void addPipeline(ChannelHandler ch) {
		List<Object> l = new ArrayList<Object>(1);
		l.add(ch);
		setPipelines(l);
	}

	public void addPipeline(CommonPipelineBuilder<ChannelHandler> ch) {
		List<Object> l = new ArrayList<Object>(1);
		l.add(ch);
		setPipelines(l);
	}

	public void setPipelines(List<Object> pblist) {
		CommonPipelineBuilderChain<ChannelHandler> chain = null;
		if (this.pipelineBuilder != null) {
			if (this.pipelineBuilder instanceof CommonPipelineBuilderChain) {
				chain = (CommonPipelineBuilderChain<ChannelHandler>) this.pipelineBuilder;
			} else {
				chain = new CommonPipelineBuilderChain<ChannelHandler>();
				chain.add(this.pipelineBuilder);
			}
		} else {
			chain = new CommonPipelineBuilderChain<ChannelHandler>();
		}
		this.pipelineBuilder = chain;

		for (Object obj : pblist) {
			if (obj instanceof CommonPipelineBuilder) {
				CommonPipelineBuilder<ChannelHandler> pb = (CommonPipelineBuilder<ChannelHandler>) obj;
				chain.add(pb);
			} else if (obj instanceof ChannelHandler) {
				ChannelHandler ch = (ChannelHandler) obj;
				chain.addHandler(
						Integer.toHexString(System.identityHashCode(ch)), ch);
			} else if (obj instanceof NamedChannelHandler) {
				NamedChannelHandler ch = (NamedChannelHandler) obj;
				chain.addHandler(ch.getName(), ch.getHandler());
			} else {
				throw new IllegalArgumentException("unknow object type - "
						+ obj.getClass().getName());
			}

		}
	}

	public Channel getServerChannel() {
		return serverChannel;
	}

	public ChannelGroup getChannelGroup() {
		return channelGroup;
	}

	public ServerBootstrap getBootstrap() {
		return bootstrap;
	}

	protected void beforeBuildPipeline(ChannelPipeline pipeline) {
		pipeline.addFirst("group", new ChannelHandlerGroup(channelGroup));
		if (traceBufferSize > 0) {
			pipeline.addLast("uplog", new ChannelHandlerLog(log, LEVEL.DEBUG,
					TYPE.UPSTREAM, traceBufferSize));
		} else {
			pipeline.addLast("showConnect", new ChannelHandlerShowConnect(log,
					LEVEL.DEBUG));
		}
	}

	protected void afterBuildPipeline(ChannelPipeline pipeline) {
		if (traceBufferSize > 0) {
			pipeline.addLast("downlog", new ChannelHandlerLog(log, LEVEL.DEBUG,
					TYPE.DOWNSTREAM, traceBufferSize));
		}
	}

	public void initServer() {
		if (this.hostPort == null)
			throw new NullPointerException("hostPort");
		if (channelGroup == null) {
			channelGroup = new DefaultChannelGroup(name);			
		}
		NettySquare.add(getName(), channelGroup);
		if (bossExecutor == null) {
			ownBossExecutor = true;
			bossExecutor = Executors.newFixedThreadPool(1);
			ThreadPoolSquare.add(name + " I/O boss", bossExecutor);
		}
		if (workerExecutor == null) {
			ownWorkerExecutor = true;
			int sz = worker;
			workerExecutor = Executors.newFixedThreadPool(sz);
			ThreadPoolSquare.add(name + " I/O worker", workerExecutor);
		}
		if (bootstrap == null) {
			bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
					bossExecutor, workerExecutor, worker));
			if (noDelay != null) {
				bootstrap.setOption("child.tcpNoDelay", noDelay.booleanValue());
			}

			bootstrap.setPipelineFactory(new ChannelPipelineFactory() {

				@Override
				public ChannelPipeline getPipeline() throws Exception {
					ChannelPipeline r = Channels.pipeline();
					beforeBuildPipeline(r);
					if (pipelineBuilder != null) {
						pipelineBuilder.buildPipeline(new NettyChannelPipeline(
								r));
					}
					afterBuildPipeline(r);
					return r;
				}
			});
		}
	}

	public void start() {
		if (this.hostPort == null || this.hostPort.getPort() == 0) {
			if (log.isInfoEnabled()) {
				log.info("{} skip startup", this.name);
			}
			return;
		}

		AIStackSimple<Boolean> stack = new AIStackSimple<Boolean>(null);
		startServer(stack);
		try {
			stack.get();
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		} catch (ExecutionException e) {
			throw ExceptionUtil.throwRuntime(e.getCause());
		}
	}

	@Override
	public void startServer() {
		start();
	}

	public boolean startServer(AIStack<Boolean> stack) {
		initServer();
		if (serverChannel == null) {
			try {
				if (log.isInfoEnabled()) {
					log.info("{} bind {}", this.name, this.hostPort);
				}
				serverChannel = bootstrap.bind(this.hostPort
						.createInetSocketAddress());
				stack.success(true);
			} catch (Exception e) {
				if (log.isWarnEnabled()) {
					log.warn(this.name + " bind " + this.hostPort + " - fail",
							e);
				}
				stack.failure(e);
			}
		}
		stack.success(true);
		return true;
	}

	public void close() {
		AIStackSimple<Boolean> stack = new AIStackSimple<Boolean>(null);
		closeServer(stack);
		try {
			stack.get(10, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		} catch (ExecutionException e) {
			throw ExceptionUtil.throwRuntime(e.getCause());
		} catch (TimeoutException e) {
		}
	}

	protected class CloseListener implements ChannelFutureListener,
			ChannelGroupFutureListener {

		private AIStack<Boolean> stack;
		private volatile boolean close1;
		private volatile boolean close2;

		public CloseListener(AIStack<Boolean> stack) {
			super();
			this.stack = stack;
		}

		public void setClose1(boolean close1) {
			this.close1 = close1;
		}

		public void setClose2(boolean close2) {
			this.close2 = close2;
		}

		public boolean isAllClose() {
			return this.close1 && this.close2;
		}

		@Override
		public void operationComplete(ChannelGroupFuture future)
				throws Exception {
			this.close2 = true;
			checkAll();
		}

		@Override
		public void operationComplete(ChannelFuture future) throws Exception {
			this.close1 = true;
			checkAll();
		}

		public void checkAll() {
			if (isAllClose()) {
				if (ownWorkerExecutor && workerExecutor != null) {
					ExecutorService s = ((ExecutorService) workerExecutor);
					if (!s.isShutdown()) {
						s.shutdown();
					}
				}
				if (ownBossExecutor && bossExecutor != null) {
					ExecutorService s = ((ExecutorService) bossExecutor);
					if (!s.isShutdown()) {
						s.shutdown();
					}
				}
				try {
					if (ownWorkerExecutor && workerExecutor != null) {
						ExecutorService s = ((ExecutorService) workerExecutor);
						if (!s.isTerminated()) {
							s.awaitTermination(1, TimeUnit.SECONDS);
						}
						ThreadPoolSquare.remove(s);
					}
				} catch (Exception e) {
				}
				try {
					if (ownBossExecutor && bossExecutor != null) {
						ExecutorService s = ((ExecutorService) bossExecutor);
						if (!s.isTerminated()) {
							s.awaitTermination(1, TimeUnit.SECONDS);
						}
						ThreadPoolSquare.remove(s);
					}
				} catch (Exception e) {
				}
				NettySquare.remove(channelGroup);
				stack.success(true);
			}
		}

	}

	@Override
	public void stopServer() {
		close();
	}

	public boolean closeServer(AIStack<Boolean> stack) {
		boolean r = false;
		CloseListener cl = new CloseListener(stack);
		if (serverChannel != null && serverChannel.isOpen()) {
			if (log.isInfoEnabled()) {
				log.info("{} close {}", this.name, this.hostPort);
			}
			serverChannel.close().addListener(cl);
			serverChannel = null;
			r = true;
		} else {
			cl.setClose1(true);
		}
		if (channelGroup != null && !channelGroup.isEmpty()) {
			channelGroup.close().addListener(cl);
		} else {
			cl.setClose2(true);
		}
		cl.checkAll();
		return r;
	}

	@Override
	public String toString() {
		return "NettyServer[" + this.name + "-" + this.hostPort + "]";
	}
}
